package hn.cch.kafka.service.impl;

import hn.cch.kafka.listener.BatchListener;
import hn.cch.kafka.service.KafkaService;
import org.apache.kafka.common.protocol.MessageUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;


@Service
public class KafkaServiceImpl implements KafkaService {

    @Autowired
    private ConcurrentKafkaListenerContainerFactory<String, String> containerFactory;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private BatchListener batchLister;

    public void send(String topic, String key, String data) {
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, key, data);

    }

    public void subscribe(String topic) {
        ConcurrentMessageListenerContainer<String, String> container = containerFactory.createContainer(topic);
        container.setupMessageListener(batchLister);
        container.start();

    }


}
